package com.edata.bigdata.flink.postgres

import org.apache.flink.connector.jdbc.JdbcStatementBuilder

import java.lang.reflect.Field
import java.sql.PreparedStatement
import scala.reflect.ClassTag


class Statement[T: ClassTag] extends JdbcStatementBuilder[Array[String]] {
  var id = ""
  var fields: Array[Field] = _
  var sinkSQL: String = _

  override def accept(t: PreparedStatement, u: Array[String]): Unit = {
    fields.zipWithIndex.foreach(f => {
      val field = f._1
      val index = f._2
      val fieldType = field.getGenericType.getTypeName
      fieldType match {
        case "java.lang.String" => t.setString(index, u(index))
        case "int" => t.setInt(index, u(index).toInt)
      }
    })
  }

  def setId(fieldName: String): Statement[T] = {
    id = fieldName
    this
  }

  def build(): Statement[T] = {
    val clazz = implicitly[ClassTag[T]].runtimeClass
    val className = clazz.getName
    val tableName = className.substring(className.lastIndexOf(".") + 1).toUpperCase()
    fields = clazz.getDeclaredFields
    val fieldNames = fields.map(f => {
      f.getName
    })
    val placeholder = fieldNames.map(fn => {
      "?"
    })
    val notIdFieldNames = fieldNames.filterNot(fieldName => fieldName.equals(id))
    val excludedField = notIdFieldNames.map(fieldName => {
      s"${fieldName} = EXCLUDED.${fieldName}"
    })

    sinkSQL =
      s"""
         |INSERT INTO ${tableName} (${fieldNames.mkString(",")}) VALUES (${placeholder.mkString(",")})
         |ON CONFLICT (${id})
         |SET ${excludedField.mkString(",")}
         |""".stripMargin
    this
  }


}
